package com.bxl.netty.rpc.client.context;

import com.bxl.netty.rpc.client.*;
import com.bxl.netty.rpc.client.common.client.AppInfo;
import com.bxl.netty.rpc.client.common.client.ClientInfo;
import com.bxl.netty.rpc.client.common.pool.ZoneClientPool;
import com.bxl.netty.rpc.client.common.register.CallBackRegister;
import com.bxl.netty.rpc.client.handler.ClientHandler;
import com.bxl.netty.rpc.client.handler.ClientMsgDecoder;
import com.bxl.netty.rpc.client.handler.ClientMsgEncoder;
import com.bxl.netty.rpc.client.common.rule.PollLoadBalance;
import com.bxl.netty.rpc.common.Lifecycle;
import com.bxl.netty.rpc.common.builder.MsgBuilder;
import com.bxl.netty.rpc.common.context.Context;
import com.bxl.netty.rpc.common.future.CompletableFuture;
import com.bxl.netty.rpc.common.message.BaseResponse;
import com.bxl.netty.rpc.common.message.PackageMsg;
import com.bxl.netty.rpc.common.message.RegisterContent;
import com.bxl.netty.rpc.common.pool.ThreadPoolUtil;
import com.bxl.netty.rpc.common.serialize.KryoSerialize;
import com.bxl.netty.rpc.common.serialize.Serialize;
import com.bxl.netty.rpc.common.util.HandlerFactory;
import com.bxl.netty.rpc.register.client.RegisterClientContext;
import com.bxl.netty.rpc.register.client.RegisterClientContextImpl;
import io.netty.channel.socket.nio.NioSocketChannel;


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * <>
 *
 * @author baixl
 * @date 2021/2/26
 */
public class ClientContextImpl implements ClientContext {
    private Lifecycle client;
    private Serialize serialize;
    private ZoneClientPool clientPool;
    private RpcClientFactory clientFactory;
    private RegisterClientContext registerClient;
    private List<HandlerFactory> handler;

    private List<HandlerFactory> decoder;
    private List<HandlerFactory> encoder;

    public RpcClientFactory getClientFactory() {
        return clientFactory;
    }

    public void setClientFactory(RpcClientFactory clientFactory) {
        this.clientFactory = clientFactory;
    }

    public ZoneClientPool getClientPool() {
        return clientPool;
    }

    public void setClientPool(ZoneClientPool clientPool) {
        this.clientPool = clientPool;
    }

    public Context init() {
        configRegister();

        configSerialize();
        configClientFactory();
        configHandler();
        configDecoder();
        configEncoder();
        configClientPool();
        return this;
    }

    protected void configRegister() {
        this.registerClient = new RegisterClientContextImpl();
        this.registerClient.init();
        this.clientPool = new ZoneClientPool();

        AppInfo zoneInfo = new AppInfo();
        zoneInfo.setApp("register");
        List<ClientInfo> clients = new ArrayList<ClientInfo>();
        //可以添加多个注册中心
        clients.add(new ClientInfo("register","localhost", 7777, this.registerClient));
        zoneInfo.setClients(clients);
        zoneInfo.setLoadBalance(new PollLoadBalance());
        this.clientPool.addZone(zoneInfo);

    }

    protected void configDecoder() {
        decoder = new ArrayList<HandlerFactory>();
        decoder.add(new HandlerFactory<Context>(ClientMsgDecoder.class));
    }

    protected void configEncoder() {
        encoder = new ArrayList<HandlerFactory>();
        encoder.add(new HandlerFactory<Context>(ClientMsgEncoder.class));
    }

    protected void configHandler() {
        handler = new ArrayList<HandlerFactory>();
        handler.add(new HandlerFactory<Context>(ClientHandler.class));
    }

    protected void configClientFactory() {
        this.clientFactory = new LocalRpcClientFactory(this);
    }

    public List<HandlerFactory> getHandler() {
        return handler;
    }

    public void setHandler(List<HandlerFactory> handler) {
        this.handler = handler;
    }

    protected void configClientPool() {
        //先初始化register-client，传入register的contextimpl
        if (this.clientPool == null) {
            this.clientPool = new ZoneClientPool();
        }


        //从注册中心获取服务信息 多个zoneid 需要初始化多个
        NioSocketChannel register = this.clientPool.getClientInfo("register").getClient();
        PackageMsg<RegisterContent> getMsg = MsgBuilder.createMsg(3, "localhost", "", 0);
        CompletableFuture<BaseResponse> cf = new CompletableFuture<BaseResponse>();
        CallBackRegister.register(getMsg.getHeader().getRequestID(), cf);
        register.writeAndFlush(getMsg);
        BaseResponse response = null;
        try {
            response = cf.get();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
        AppInfo zoneInfo = (AppInfo) response.getData();
        List<ClientInfo> clients = zoneInfo.getClients();
        for (ClientInfo clientInfo : clients) {
            clientInfo.setContext(this);
        }
        zoneInfo.setLoadBalance(new PollLoadBalance());
        clientPool.addZone(zoneInfo);
        clientPool.init();

        ThreadPoolUtil.scheduleAtFixedRate(new GetClientsTask(clientPool, getMsg, this), 40, 40, TimeUnit.SECONDS);
    }


    private static class GetClientsTask implements Runnable {
        private ZoneClientPool clientPool;
        private PackageMsg<RegisterContent> getMsg;
        private ClientContext context;
        private int pullTime = 0;
        List<ClientInfo> newClients;

        public GetClientsTask(ZoneClientPool clientPool, PackageMsg<RegisterContent> getMsg, Context context) {
            this.clientPool = clientPool;
            this.getMsg = getMsg;
            if (context instanceof ClientContext) {
                this.context = (ClientContext) context;
            }
        }

        public void run() {
            NioSocketChannel register = this.clientPool.getClientInfo("register").getClient();
            CompletableFuture<BaseResponse> cf = new CompletableFuture<BaseResponse>();
            CallBackRegister.register(getMsg.getHeader().getRequestID(), cf);
            register.writeAndFlush(getMsg);
            BaseResponse response = null;
            try {
                response = cf.get();
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
            if (response == null) {
                return;
            }
            AppInfo zoneInfo = (AppInfo) response.getData();
            //拉取的client
            List<ClientInfo> syncClients = zoneInfo.getClients();
            //拉取的服务为空，可能是注册中心出了问题，暂不清除客户端
            if (syncClients == null || syncClients.isEmpty()) {
                pullTime++;
                return;
            }
            //存在的client
                List<ClientInfo> newClients = zoneInfo.getOldClients();
                newClients.clear();
            List<ClientInfo> existsClients = context.getClientPool().getClients(getMsg.getContent().getApp());
            for (ClientInfo syncClient : syncClients) {
                if (!existsClients.contains(syncClient)) {
                    syncClient.setContext(context);
                    newClients.add(syncClient);
                }
            }
            for (ClientInfo existsClient : existsClients) {
                if (syncClients.contains(existsClient)) {
                    newClients.add(existsClient);
                } else {
                    existsClient.stop();
                }

            }
            context.getClientPool().setClients(getMsg.getContent().getApp(), newClients);
            zoneInfo.setOldClients(existsClients);
        }
    }

    /**
     * 可以修改为配置化
     */
    protected void configSerialize() {
        serialize = new KryoSerialize();
    }


    public Serialize getSerialize() {
        return serialize;
    }

    public List<HandlerFactory> getDecoder() {
        return decoder;
    }

    public List<HandlerFactory> getEncoder() {
        return encoder;
    }
}
